/**
 * 
 */
package com.xpec.c4.service.group.datalayer.infinispan;

import java.util.Collection;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.infinispan.Cache;

import com.xpec.c4.codec.marshalling.MarshallingCodec;
import com.xpec.c4.service.group.datalayer.IShareMap;
import com.xpec.c4.service.module.ServerProperties;

/**
 * server组内共享的Map<br>
 * 在Map内的数据，组内可以完全共享
 * 
 * @author lipeilin
 * @2011-1-25 下午03:32:23
 * 
 */
public class InfinispanShareMap<K, V> implements IShareMap<K, V> {

	private static final Log log = LogFactory.getLog(InfinispanShareMap.class);

	/**
	 * 是否进行分析？
	 */
	private static final boolean sharemapStatEnable = ServerProperties
			.getBoolean("sharemapStatEnable", false);

	/**
	 * 分析的频率，多久处理一次
	 */
	private static final int sharemapStatRate = ServerProperties.getInt(
			"sharemapStatRate", 100);

	/**
	 * 上次的统计时间
	 */
	private long stateLasttime = System.currentTimeMillis();

	/**
	 * 上次至今的统计次数
	 */
	private int stateCount = 0;

	/**
	 * 共享容器
	 */
	private Cache<K, V> cache;

	/**
	 * 唯一标识别名
	 */
	private String name;

	/**
	 * marshalling编码器
	 */
	private static MarshallingCodec marshallingCodec = new MarshallingCodec();

	/**
	 * 
	 */
	InfinispanShareMap(Cache<K, V> cahche, String name) {
		this.cache = cahche;
		this.name = name;
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see java.util.Map#clear()
	 */
	/*
	 * (non-Javadoc)
	 * 
	 * @see com.xpec.c4.service.group.infinispan.NonShareMap#clear()
	 */
	@Override
	public void clear() {
		cache.clear();

	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see java.util.Map#containsKey(java.lang.Object)
	 */
	/*
	 * (non-Javadoc)
	 * 
	 * @see
	 * com.xpec.c4.service.group.infinispan.NonShareMap#containsKey(java.lang
	 * .Object)
	 */
	@Override
	public boolean containsKey(Object key) {

		if (key == null) {
			return false;
		}

		// 统计
		if (sharemapStatEnable) {

			stateCount++;

			// 如果够次数了，就统计显示信息
			if (stateCount % sharemapStatRate == 0) {

				long now = stateLasttime;
				stateLasttime = System.currentTimeMillis();
				long time = stateLasttime - now;

				// 本次的处理时间
				boolean result = cache.containsKey(key);
				long dealTime = System.currentTimeMillis() - stateLasttime ;

				log.warn("ShareMap " + this.name + " containsKey " + " key:"
						+ key + " totalCount:" + stateCount + " rateTime:"
						+ time + " rateAvg:" + (time / sharemapStatRate)
						+ " dealTime:" + dealTime);
				log.warn("ShareMap " + this.name
						+ " containsKey Marshalling length:"
						+ getMarshallingLength(key) + " Object:" + key);

				return result;

			}

		}

		return cache.containsKey(key);
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see java.util.Map#containsValue(java.lang.Object)
	 */
	/*
	 * (non-Javadoc)
	 * 
	 * @see
	 * com.xpec.c4.service.group.infinispan.NonShareMap#containsValue(java.lang
	 * .Object)
	 */
	@Override
	public boolean containsValue(Object value) {

		// 统计
		if (sharemapStatEnable) {

			stateCount++;

			// 如果够次数了，就统计显示信息
			if (stateCount % sharemapStatRate == 0) {
				long now = stateLasttime;
				stateLasttime = System.currentTimeMillis();
				long time = stateLasttime - now;

				// 本次的处理时间
				boolean result = cache.containsValue(value);
				long dealTime = System.currentTimeMillis() - stateLasttime ;

				log.warn("ShareMap " + this.name + " containsValue "
						+ " value:" + value + " totalCount:" + stateCount
						+ " time:" + time + " avg:" + (time / 10)
						+ " dealTime:" + dealTime);
				log.warn("ShareMap " + this.name
						+ " containsValue Marshalling length:"
						+ getMarshallingLength(value) + " Object:" + value);

				return result;

			}
		}

		return cache.containsValue(value);
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see java.util.Map#entrySet()
	 */
	/*
	 * (non-Javadoc)
	 * 
	 * @see com.xpec.c4.service.group.infinispan.NonShareMap#entrySet()
	 */
	@Override
	public Set<Map.Entry<K, V>> entrySet() {

		if (sharemapStatEnable) {
			log.warn("ShareMap entrySet " + this.name);
		}

		return cache.entrySet();
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see java.util.Map#get(java.lang.Object)
	 */
	/*
	 * (non-Javadoc)
	 * 
	 * @see
	 * com.xpec.c4.service.group.infinispan.NonShareMap#get(java.lang.Object)
	 */
	@Override
	public V get(Object key) {
		if (key == null) {
			return null;
		}

		// 统计
		if (sharemapStatEnable) {

			stateCount++;

			// 如果够次数了，就统计显示信息
			if (stateCount % sharemapStatRate == 0) {
				long now = stateLasttime;
				stateLasttime = System.currentTimeMillis();
				long time = stateLasttime - now;

				// 本次的处理时间
				V value = cache.get(key);
				long dealTime = System.currentTimeMillis() - stateLasttime ;

				log.warn("ShareMap " + this.name + " get " + " key:" + key
						+ " totalCount:" + stateCount + " time:" + time
						+ " avg:" + (time / 10) + " dealTime:" + dealTime);
				log.warn("ShareMap " + this.name + " get Marshalling length:"
						+ getMarshallingLength(key) + " Object:" + key);

				return value;

			}
		}

		// log.info("sharemap get ! " + name + " key:" + key + " begin!");
		// long begin = System.currentTimeMillis();
		V value = cache.get(key);
		// log.info(Thread.currentThread()+" get from sharemap "+name+" key:"+key+" value:"+value+" end! time:"+(System.currentTimeMillis()-begin));

		return value;
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see java.util.Map#isEmpty()
	 */
	/*
	 * (non-Javadoc)
	 * 
	 * @see com.xpec.c4.service.group.infinispan.NonShareMap#isEmpty()
	 */
	@Override
	public boolean isEmpty() {
		return cache.isEmpty();
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see java.util.Map#keySet()
	 */
	/*
	 * (non-Javadoc)
	 * 
	 * @see com.xpec.c4.service.group.infinispan.NonShareMap#keySet()
	 */
	@Override
	public Set<K> keySet() {
		return cache.keySet();
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see java.util.Map#put(java.lang.Object, java.lang.Object)
	 */
	/*
	 * (non-Javadoc)
	 * 
	 * @see com.xpec.c4.service.group.infinispan.NonShareMap#put(K, V)
	 */
	@Override
	public synchronized V put(K key, V value) {
		putAsync(key, value);
		return value;
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see java.util.Map#putAll(java.util.Map)
	 */
	/*
	 * (non-Javadoc)
	 * 
	 * @see
	 * com.xpec.c4.service.group.infinispan.NonShareMap#putAll(java.util.Map)
	 */
	@Override
	public void putAll(Map<? extends K, ? extends V> m) {

		// 统计
		if (sharemapStatEnable) {

			stateCount++;

			// 如果够次数了，就统计显示信息
			if (stateCount % sharemapStatRate == 0) {
				long now = stateLasttime;
				stateLasttime = System.currentTimeMillis();
				long time = stateLasttime - now;

				// 本次的处理时间
				cache.putAll(m);
				long dealTime = System.currentTimeMillis() - stateLasttime ;

				log.warn("ShareMap " + this.name + " putAll " + " Map:" + m
						+ " totalCount:" + stateCount + " time:" + time
						+ " avg:" + (time / 10) + " dealTime:" + dealTime);

				for (Entry<? extends K, ? extends V> entry : m.entrySet()) {
					log.warn("ShareMap " + this.name
							+ " putAll key Marshalling length:"
							+ getMarshallingLength(entry.getKey()) + " Object:"
							+ entry.getKey());
					log.warn("ShareMap " + this.name
							+ " putAll value Marshalling length:"
							+ getMarshallingLength(entry.getValue())
							+ " Object:" + entry.getValue());
				}
				return;

			}
		}

		cache.putAll(m);

	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see java.util.Map#remove(java.lang.Object)
	 */
	/*
	 * (non-Javadoc)
	 * 
	 * @see
	 * com.xpec.c4.service.group.infinispan.NonShareMap#remove(java.lang.Object)
	 */
	@Override
	public synchronized V remove(Object key) {
		removeAsync((K) key);
		return null;
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see com.xpec.c4.service.group.infinispan.NonShareMap#size()
	 */
	@Override
	@Deprecated
	public int size() {
		return cache.size();
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see java.util.Map#values()
	 */
	/*
	 * (non-Javadoc)
	 * 
	 * @see com.xpec.c4.service.group.infinispan.NonShareMap#values()
	 */
	@Override
	public Collection<V> values() {
		return cache.values();
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see com.xpec.c4.service.group.infinispan.NonShareMap#getName()
	 */
	public String getName() {
		return name;
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see com.xpec.c4.service.group.infinispan.NonShareMap#totalSize()
	 */
	@Deprecated
	public int totalSize() {
		// TODO 全部数量，还没实现
		return entrySet().size();
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see com.xpec.c4.service.group.infinispan.NonShareMap#localSize()
	 */
	public int localSize() {
		return cache.size();
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see com.xpec.c4.service.group.infinispan.NonShareMap#lock(K)
	 */
	public boolean lock(K... keys) {
		return false;
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see com.xpec.c4.service.group.infinispan.NonShareMap#getCache()
	 */
	public Cache<K, V> getCache() {
		return cache;
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see com.xpec.c4.service.group.infinispan.NonShareMap#txBegin()
	 */
	public void txBegin() {
		throw new RuntimeException("not use!");
		// try {
		// cache.getCacheManager().getTransactionManager().begin();
		// } catch (NotSupportedException e) {
		// log.error(cache, e);
		// } catch (SystemException e) {
		// log.error(cache, e);
		// }
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see com.xpec.c4.service.group.infinispan.NonShareMap#txCommit()
	 */
	public void txCommit() {
		throw new RuntimeException("not use!");
		// try {
		// cache.getTransactionManager().commit();
		// } catch (SecurityException e) {
		// log.error(cache, e);
		// } catch (IllegalStateException e) {
		// log.error(cache, e);
		// } catch (RollbackException e) {
		// log.error(cache, e);
		// } catch (HeuristicMixedException e) {
		// log.error(cache, e);
		// } catch (HeuristicRollbackException e) {
		// log.error(cache, e);
		// } catch (SystemException e) {
		// log.error(cache, e);
		// }
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see com.xpec.c4.service.group.infinispan.NonShareMap#txRollback()
	 */
	public void txRollback() {
		throw new RuntimeException("not use!");
		// try {
		// cache.getTransactionManager().rollback();
		// } catch (IllegalStateException e) {
		// log.error(cache, e);
		// } catch (SecurityException e) {
		// log.error(cache, e);
		// } catch (SystemException e) {
		// log.error(cache, e);
		// }
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see
	 * com.xpec.c4.service.group.datalayer.IShareMap#delete(java.lang.Object)
	 */
	@Override
	public V delete(Object key) {
		return remove(key);
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see
	 * com.xpec.c4.service.group.datalayer.IShareMap#putForExternalRead(java
	 * .lang.Object, java.lang.Object)
	 */
	@Override
	public void putForExternalRead(K key, V value) {
		if (log.isDebugEnabled()) {
			log.debug(Thread.currentThread()
					+ " putForExternalRead to sharemap " + name + " key:" + key
					+ " value:" + value);
		}

		// 统计
		if (sharemapStatEnable) {

			stateCount++;

			// 如果够次数了，就统计显示信息
			if (stateCount % sharemapStatRate == 0) {
				long now = stateLasttime;
				stateLasttime = System.currentTimeMillis();
				long time = stateLasttime - now;

				// 本次的处理时间
				cache.putForExternalRead(key, value);
				long dealTime = System.currentTimeMillis() - stateLasttime ;

				log.warn("ShareMap " + this.name + " putForExternalRead "
						+ " key:" + key + " value:" + value + " totalCount:"
						+ stateCount + " time:" + time + " avg:" + (time / 10)
						+ " dealTime:" + dealTime);
				log.warn("ShareMap " + this.name
						+ " putForExternalRead Marshalling length:"
						+ getMarshallingLength(key) + " Object:" + key);
				log.warn("ShareMap " + this.name
						+ " putForExternalRead Marshalling length:"
						+ getMarshallingLength(value) + " Object:" + value);

				return;

			}
		}

		// log.info("sharemap putForExternalRead ! " + name + " key:" + key
		// + " value:" + value + " begin!");
		// txBegin();
		cache.putForExternalRead(key, value);
		// txCommit();

	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see com.xpec.c4.service.group.datalayer.IShareMap#put(java.lang.Object,
	 * java.lang.Object, long)
	 */
	@Override
	public void put(K key, V value, long lifespan) {

		// 统计
		if (sharemapStatEnable) {

			stateCount++;

			// 如果够次数了，就统计显示信息
			if (stateCount % sharemapStatRate == 0) {
				long now = stateLasttime;
				stateLasttime = System.currentTimeMillis();
				long time = stateLasttime - now;

				// 本次的处理时间
				cache.put(key, value, lifespan, TimeUnit.MILLISECONDS);
				long dealTime = System.currentTimeMillis() - stateLasttime ;

				log.warn("ShareMap " + this.name + " lifespan put " + " key:"
						+ key + " value:" + value + " totalCount:" + stateCount
						+ " time:" + time + " avg:" + (time / 10)
						+ " dealTime:" + dealTime);
				log.warn("ShareMap " + this.name
						+ " putSync Marshalling length:"
						+ getMarshallingLength(key) + " Object:" + key);
				log.warn("ShareMap " + this.name
						+ " putSync Marshalling length:"
						+ getMarshallingLength(value) + " Object:" + value);

				return;

			}
		}

		cache.put(key, value, lifespan, TimeUnit.MILLISECONDS);

	}

	public void setCache(Cache<K, V> cache) {
		this.cache = cache;
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see
	 * com.xpec.c4.service.group.datalayer.IShareMap#putAsyn(java.lang.Object,
	 * java.lang.Object)
	 */
	@Override
	public Future<V> putAsync(K key, V value) {
		if (key == null) {
			log.error("sharemap put key is null! key:" + key + " name:"
					+ this.name);
			return null;
		}
		if (value == null) {
			log.error("sharemap put value is null! value:" + value + " name:"
					+ this.name);
			return null;
		}

		// 统计
		if (sharemapStatEnable) {

			stateCount++;

			// 如果够次数了，就统计显示信息
			if (stateCount % sharemapStatRate == 0) {
				long now = stateLasttime;
				stateLasttime = System.currentTimeMillis();
				long time = stateLasttime - now;

				// 本次的处理时间
				Future<V> result = cache.putAsync(key, value);
				long dealTime = System.currentTimeMillis() - stateLasttime ;

				log.warn("ShareMap " + this.name + " putAsync " + " key:" + key
						+ " value:" + value + " totalCount:" + stateCount
						+ " time:" + time + " avg:" + (time / 10)
						+ " dealTime:" + dealTime);
				log.warn("ShareMap putAsync Marshalling length:"
						+ getMarshallingLength(key) + " Object:" + key);
				log.warn("ShareMap putAsync Marshalling length:"
						+ getMarshallingLength(value) + " Object:" + value);

				return result;
			}
		}

		// log.info("sharemap putAsync ! " + name + " key:" + key + " value:"
		// + value + " begin!");
		// long begin = System.currentTimeMillis();
		Future<V> result = cache.putAsync(key, value);
		// log.info(Thread.currentThread()+" putAsync to sharemap "+name+" key:"+key+" value:"+value+" end! time:"+(System.currentTimeMillis()-begin));
		return result;
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see
	 * com.xpec.c4.service.group.datalayer.IShareMap#removeAsyn(java.lang.Object
	 * )
	 */
	@Override
	public Future<V> removeAsync(K key) {
		if (key == null) {
			log.warn("sharemap removeAsync key is null! name:" + this.name);
			return null;
		}

		// 统计
		if (sharemapStatEnable) {

			stateCount++;

			// 如果够次数了，就统计显示信息
			if (stateCount % sharemapStatRate == 0) {
				long now = stateLasttime;
				stateLasttime = System.currentTimeMillis();
				long time = stateLasttime - now;

				// 本次的处理时间
				Future<V> result = cache.removeAsync(key);
				long dealTime = System.currentTimeMillis() - stateLasttime ;

				log.warn("ShareMap " + this.name + " removeAsync " + " key:"
						+ key + " totalCount:" + stateCount + " time:" + time
						+ " avg:" + (time / 10) + " dealTime:" + dealTime);
				log.warn("ShareMap removeAsync Marshalling length:"
						+ getMarshallingLength(key) + " Object:" + key);
				return result;

			}
		}

		// log.info("sharemap removeAsync ! " + name + " key:" + key +
		// " begin!");
		// long begin = System.currentTimeMillis();
		Future<V> result = cache.removeAsync(key);
		// log.info(Thread.currentThread()+" removeAsync to sharemap "+name+" key:"+key+" end! time:"+(System.currentTimeMillis()-begin));
		return result;
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see
	 * com.xpec.c4.service.group.datalayer.IShareMap#putAsyn(java.lang.Object,
	 * java.lang.Object)
	 */
	@Override
	public V putSync(K key, V value) {
		if (key == null) {
			log.error("sharemap put key is null! value:" + value + " name:"
					+ this.name);
			return null;
		}
		if (value == null) {
			log.error("sharemap put value is null! key:" + key + " name:"
					+ this.name);
			return null;
		}

		// 统计
		if (sharemapStatEnable) {

			stateCount++;

			// 如果够次数了，就统计显示信息
			if (true) {
				long now = stateLasttime;
				stateLasttime = System.currentTimeMillis();
				long time = stateLasttime - now;

				// 本次的处理时间
				V result = cache.put(key, value);
				long dealTime = System.currentTimeMillis() - stateLasttime ;

				log.warn("ShareMap " + this.name + " putSync " + " key:" + key
						+ " value:" + value + " totalCount:" + stateCount
						+ " time:" + time + " avg:" + (time / 10)
						+ " dealTime:" + dealTime);
				log.warn("ShareMap " + this.name
						+ " putSync Marshalling length:"
						+ getMarshallingLength(key) + " Object:" + key);
				log.warn("ShareMap " + this.name
						+ " putSync Marshalling length:"
						+ getMarshallingLength(value) + " Object:" + value);
				return result;

			}
		}

		// log.info("sharemap putSync! " + name + " key:" + key + " begin!");
		return cache.put(key, value);
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see
	 * com.xpec.c4.service.group.datalayer.IShareMap#removeAsyn(java.lang.Object
	 * )
	 */
	@Override
	public V removeSync(K key) {
		if (key == null) {
			log.warn("sharemap removeAsync key is null! name:" + this.name);
			return null;
		}

		// 统计
		if (sharemapStatEnable) {

			stateCount++;

			// 如果够次数了，就统计显示信息
			if (stateCount % sharemapStatRate == 0) {
				long now = stateLasttime;
				stateLasttime = System.currentTimeMillis();
				long time = stateLasttime - now;

				// 本次的处理时间
				V result = cache.remove(key);
				long dealTime = System.currentTimeMillis() - stateLasttime ;

				log.warn("ShareMap " + this.name + " removeSync " + " key:"
						+ key + " totalCount:" + stateCount + " time:" + time
						+ " avg:" + (time / 10) + " dealTime:" + dealTime);
				log.warn("ShareMap " + this.name
						+ " putSync Marshalling length:"
						+ getMarshallingLength(key) + " Object:" + key);
				return result;

			}
		}

		// log.error("sharemap removeSync! key:" + this.name + " key:" + key);
		return cache.remove(key);
	}

	private int getMarshallingLength(Object obj) {
		try {
			return marshallingCodec.encodeByte(obj).length;
		} catch (Exception e) {
			log.error("marshallingCodec error! " + obj, e);
		}
		return 0;
	}

}
